/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.gitee.kamismile.stone.rpc.grpc;

import com.gitee.kamismile.stone.commmon.base.ResultVO;
import com.gitee.kamismile.stone.commmon.exception.BusinessException;
import com.gitee.kamismile.stone.commmon.util.JsonUtil;
import com.gitee.kamismile.stone.commmon.util.ValueUtils;
import com.gitee.kamismile.stone.rpc.api.IProtocol;
import com.gitee.kamismile.stone.rpc.api.invocation.support.StoneInvocation;
import com.gitee.kamismile.stone.rpc.common.dto.StoneAppResponse;
import com.gitee.kamismile.stone.rpc.common.errorenum.RpcErrorEnum;
import com.gitee.kamismile.stone.rpc.common.exception.RPCException;
import com.gitee.kamismile.stone.rpc.common.support.StoneURL;
import com.gitee.kamismile.stone.rpc.common.utils.StoneBeanUtilsApplication;
import io.grpc.*;
import io.grpc.netty.NettyServerBuilder;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang.StringUtils;
import org.apache.commons.lang.exception.ExceptionUtils;
import  org.slf4j.LoggerFactory;
import org.slf4j.Logger;
import org.springframework.scheduling.concurrent.CustomizableThreadFactory;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.Map;
import java.util.concurrent.*;

/**
 *
 */
public class GrpcProtocol<T> implements IProtocol {

    protected final Logger logger = LoggerFactory.getLogger(GrpcProtocol.class);


    /* <address, gRPC channels> */
    private final ConcurrentMap<String, ReferenceCountManagedChannel> channelMap = new ConcurrentHashMap<>();


    private final Object lock = new Object();

    @Override
    public Runnable export(StoneInvocation invocation) {
        StoneURL url = invocation.getStoneURL();

        if (url.getPort() < 1) {
            return null;
        }

//        StoneHandlerRegistry registry = new StoneHandlerRegistry();



        NettyServerBuilder builder =
                NettyServerBuilder
                        .forPort(url.getPort())
                        .intercept(new GlobalGrpcExceptionHandler());
//                        .fallbackHandlerRegistry(registry);

        configPool(url, builder);


        Map<String, BindableService> mapBeans = StoneBeanUtilsApplication.getApplicationContext()
                .getBeansOfType(BindableService.class);

        if (MapUtils.isEmpty(mapBeans)) {
            return null;
        }

        for (BindableService bean : mapBeans.values()) {
            try {
                Class<? extends BindableService> originalImpl = bean.getClass();
                Method[] methods = originalImpl.getMethods();
                Method method = null;
                for (Method methodTmp : methods) {
                    if (methodTmp.getName().equals("setProxiedImpl")) {
                        method = methodTmp;
                        break;
                    }
                }
                method.invoke(bean, bean);
            } catch (Exception e) {
                throw new IllegalStateException(
                        "Failed to set stone proxied service impl to stub, please make sure your stub " +
                                "was generated by the stone-protoc-compiler.", e);
            }
//            registry.addService(bean);
            builder.addService(bean.bindService());
        }

        Server grpcServer = GrpcOptionsUtils.buildServerBuilder(url, builder).build();

        return () -> {
            try {
                grpcServer.start();
            } catch (Exception e) {
                logger.error(ExceptionUtils.getFullStackTrace(e));
                throw new RPCException(RpcErrorEnum.STARTSYSTEM.getCode(),
                        RpcErrorEnum.STARTSYSTEM.getMessage());
            }
        };
    }

    private  void configPool(StoneURL url, NettyServerBuilder builder) {
        if(StringUtils.isBlank(url.getCoreThreads())){
            return;
        }

            Integer coreThreads = ValueUtils.isIntegerNull(url.getCoreThreads(), Runtime.getRuntime().availableProcessors());
            Integer maxThreads = ValueUtils.isIntegerNull(url.getMaxThreads(), Runtime.getRuntime().availableProcessors() * 5);
            Integer maxIdleInSecond = ValueUtils.isIntegerNull(url.getMaxIdleInSecond(), 60);
            Integer maxQueueSize = ValueUtils.isIntegerNull(url.getMaxQueueSize(), 2000);


            ThreadPoolExecutor executorService = new ThreadPoolExecutor(
                    coreThreads,
                    maxThreads,
                    maxIdleInSecond, TimeUnit.MILLISECONDS,
                    new ArrayBlockingQueue<>(maxQueueSize),
                    new CustomizableThreadFactory("GrpcServer"));

            builder.executor(executorService) ;
    }

    @Override
    public StoneAppResponse refer(Class type, StoneURL url) {

        Class<?> enclosingClass = type.getEnclosingClass();

        if (enclosingClass == null) {
            throw new IllegalArgumentException(
                    type.getName() + " must be declared inside protobuf generated classes, " +
                            "should be something like ServiceNameGrpc.IServiceName.");
        }

        final Method stoneStubMethod;
        try {
            stoneStubMethod = enclosingClass.getDeclaredMethod("getStoneStub", Channel.class,
                    CallOptions.class,
                    StoneURL.class);
        } catch (NoSuchMethodException e) {
            throw new IllegalArgumentException("Does not find getStoneStub in " + enclosingClass.getName()
                    + ", please use the customized protoc-gen-Stone-java to update the generated classes.");
        }
        ReferenceCountManagedChannel channel = getSharedChannel(url);
        try {
            @SuppressWarnings("unchecked") final T proxy = (T) stoneStubMethod.invoke(null,
                    channel,
                    GrpcOptionsUtils.buildCallOptions(url),
                    url
            );

            StoneAppResponse stoneAppResponse = new StoneAppResponse();
            Method method = proxy.getClass().getMethod(url.getMethodName(), url.getParameterTypes());
            Object data = method.invoke(proxy, url.getArguments());
            stoneAppResponse.setData(data);

            return stoneAppResponse;
        } catch (Exception e) {
            logger.error(ExceptionUtils.getFullStackTrace(e));
            if (e instanceof InvocationTargetException) {
                Throwable targetException = ((InvocationTargetException) e).getTargetException();
                if (targetException instanceof StatusRuntimeException) {
                    StatusRuntimeException statusRuntimeException = (StatusRuntimeException) targetException;
                    Status status = statusRuntimeException.getStatus();
                    if (status.getCode() == Status.INTERNAL.getCode()) {
                        String description = status.getDescription();
                        //todo 判断 是否json
                        try {
                            ResultVO resultVO = JsonUtil.fromJson(description, ResultVO.class);
                            throw new BusinessException(Integer.parseInt(resultVO.getCode()),
                                    resultVO.getMessage(), resultVO.getData());
                        } catch (Exception ex) {
                            if (ex instanceof BusinessException) {
                                throw ex;
                            }
                        }
                    } else if (status.getCode() == Status.DEADLINE_EXCEEDED.getCode()) {
                        throw new RPCException(RpcErrorEnum.OUTTIME.getCode(),
                                "rpc out time ");
                    } else {
                        //todo 判断异常状态 抛异常
                        throw new RPCException(RpcErrorEnum.SENDSERVER.getCode(),
                                "rpc runtime error");
                    }
                }
            }
            //todo 判断异常状态 抛异常
            throw new RPCException(RpcErrorEnum.CREATESERVER.getCode(),
                    "Could not create stub through reflection.");
        }
    }

    @Override
    public void start(Runnable runnable) {
        runnable.run();
    }

    /**
     * Get shared channel connection
     */
    private ReferenceCountManagedChannel getSharedChannel(StoneURL url) {
        String key = url.getServerName() + "`" + url.getHost() + "`" + url.getPort();
        ReferenceCountManagedChannel channel = channelMap.get(key);

        if (channel != null && !channel.isTerminated()) {
            channel.incrementAndGetCount();
            return channel;
        }

        synchronized (lock) {
            channel = channelMap.get(key);
            // Stone check
            if (channel != null && !channel.isTerminated()) {
                channel.incrementAndGetCount();
            } else {
                channel = new ReferenceCountManagedChannel(initChannel(url));
                channelMap.put(key, channel);
            }
        }

        return channel;
    }

    /**
     * Create new connection
     *
     * @param url
     */
    private ManagedChannel initChannel(StoneURL url) {
        return GrpcOptionsUtils.buildManagedChannel(url);
    }


    @Override
    public void destroy() {
//        channelMap.values().forEach(ReferenceCountManagedChannel::shutdown);
//        channelMap.clear();
        //todo 不管容器 直接重置状态为关闭状态
    }


}
